package com.bw;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import java.time.Duration;
import java.util.Date;
import java.util.Properties;

/**
 * ODS层数据接入：从Kafka读取原始客户行为数据，生成ODS层数据流
 */
public class OdsShopCustomerDataLoader {
    /**
     * 加载Kafka原始数据，转换为OdsShopCustomerBehavior数据流
     * @param env Flink执行环境
     * @param kafkaBootstrapServers Kafka地址
     * @param kafkaTopic Kafka主题
     * @param kafkaGroupId 消费者组ID
     * @return ODS层数据流
     */
    public static DataStream<OdsShopCustomerBehavior> loadOdsData(
            StreamExecutionEnvironment env,
            String kafkaBootstrapServers,
            String kafkaTopic,
            String kafkaGroupId
    ) {
        // 1. 配置Kafka消费者
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", kafkaBootstrapServers);
        kafkaProps.setProperty("group.id", kafkaGroupId);
        kafkaProps.setProperty("auto.offset.reset", "latest"); // 从最新偏移量开始消费

        // 2. 读取Kafka数据并转换为ODS实体
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                kafkaTopic,
                new org.apache.flink.api.common.serialization.SimpleStringSchema(),
                kafkaProps
        );

        // 3. 反序列化+补充字段（dataDt、etlLoadTime）
        return env.addSource(kafkaConsumer)
                .map(new MapFunction<String, OdsShopCustomerBehavior>() {
                    @Override
                    public OdsShopCustomerBehavior map(String jsonStr) throws Exception {
                        // 解析JSON为ODS实体
                        OdsShopCustomerBehavior ods = JSON.parseObject(jsonStr, OdsShopCustomerBehavior.class);
                        // 补充数据日期（behaviorTime转换为yyyyMMdd）
                        ods.setDataDt(ShopCustomerConstants.DATE_FORMAT.format(ods.getBehaviorTime()));
                        // 补充ETL加载时间（当前时间）
                        ods.setEtlLoadTime(new Date());
                        return ods;
                    }
                })
                // 4. 定义Watermark（处理乱序数据，允许10秒延迟）
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OdsShopCustomerBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                                .withTimestampAssigner(new SerializableTimestampAssigner<OdsShopCustomerBehavior>() {
                                    @Override
                                    public long extractTimestamp(OdsShopCustomerBehavior element, long recordTimestamp) {
                                        return element.getBehaviorTime().getTime(); // 以行为时间为事件时间
                                    }
                                })
                );
    }
}